package study.mq.kafka.consumer.listener._1_consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * groupId     组id
 * id          线程名
 * idIsGroup   id是否作为group名
 * concurrency 并发数
 */
@Slf4j
//@Component
public class _2_ConsumerListener {

    @KafkaListener(topics = "topic-3",
            groupId = "MY_GP",
            id = "consumer_thread",
            idIsGroup = false,
            concurrency = "4")
    public void onMessage(ConsumerRecord<String, String> record, Consumer consumer) {
        //
        String topic = record.topic();
        int partition = record.partition();
        String value = record.value();

        long offset = record.offset();

        log.info("topic= {}, partition= {}, offset= {}", topic, partition, offset);
        log.info(">>>>>> {}", value);

        consumer.commitAsync();
    }
}
